
package com.bangcommunity.bbframe.mq.kafka.consumer;

import com.bangcommunity.bbframe.mq.IQtMessager;
import com.bangcommunity.bbframe.mq.consumer.IQtMsgProcessor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.Serializable;
import java.util.Arrays;

/**
 * Created by tanghc on 17/2/4.
 */
public class DelegateKafkaConsumer<T extends Serializable> implements Runnable {

    private String topic;
    private KafkaConsumer<String, IQtMessager<T>> consumer;
    private IQtMsgProcessor msgProcessor;

    public DelegateKafkaConsumer(String topic, KafkaConsumer<String, IQtMessager<T>> consumer,
            IQtMsgProcessor msgProcessor) {
        this.topic = topic;
        this.consumer = consumer;
        this.msgProcessor = msgProcessor;
    }

    @Override
    public void run() {
        if (null == consumer || msgProcessor == null) {
            return;
        }
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            ConsumerRecords<String, IQtMessager<T>> records = consumer.poll(100);
            if (records != null) {
                for (ConsumerRecord<String, IQtMessager<T>> record : records) {
                    if (null != record) {
                        IQtMessager<T> msg = record.value();
                        if (null != msg) {
                            msgProcessor.onConsume(msg.getMsgId(), msg.getBody());
                        }
                    }
                }
            }
        }
    }
}
